package org.budo.elasticsearch.jdbc.driver;

import java.net.InetAddress;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;

import org.budo.elasticsearch.jdbc.driver.parser.jsql.ElasticsearchJSqlStatementParser;
import org.budo.jdbc.driver.AbstractBudoJdbcDriver;
import org.budo.jdbc.driver.JdbcUrl;
import org.budo.jdbc.driver.parser.SqlStatementParser;
import org.budo.jdbc.driver.parser.cache.CachedJdbcStatementParser;
import org.budo.support.java.net.util.JavaNetUtil;
import org.budo.support.lang.util.ArrayUtil;
import org.budo.support.lang.util.MapUtil;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.settings.Settings.Builder;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

import lombok.extern.slf4j.Slf4j;

/**
 * jdbc:es://es_vpc.itaojin.cn:10080?cluster.name=tj-es-cluster
 * 
 * @author lmw
 * 
 * @see org.budo.elasticsearch.jdbc.driver.ESConf
 */
@Slf4j
public class BudoElasticsearchJdbcDriver extends AbstractBudoJdbcDriver {
    private static final int DEFAULT_PORT = 9300;

    private static final String[] URL_PREFIX = {"jdbc:mysql://", "jdbc:elasticsearch://", "jdbc:es://"};

    private static final SqlStatementParser STATEMENT_PARSER = new CachedJdbcStatementParser() //
            .setStatementParser(new ElasticsearchJSqlStatementParser());

    static {
        try {
            DriverManager.registerDriver(new BudoElasticsearchJdbcDriver()); // 类加载时注册驱动
        } catch (Throwable e) {
            throw new IllegalStateException("#36 Can't register driver!", e);
        }
    }

    @Override
    public boolean acceptsURL(String url) throws SQLException {
        return JdbcUrl.acceptsURL(url, URL_PREFIX);
    }

    @SuppressWarnings("resource")
    @Override
    public Connection connect(String url, Properties info) throws SQLException {
        JdbcUrl jdbcUrl=JdbcUrl.parse(url, URL_PREFIX, DEFAULT_PORT);

        TransportClient transportClient = this.transportClient(jdbcUrl.getHost(), jdbcUrl.getPort(), jdbcUrl.getParameters());

        String[] indices = transportClient.admin().indices().prepareGetIndex().get().getIndices();
        if (!ArrayUtil.contains(indices, jdbcUrl.getCatalog())) {
            log.warn("#114 indices=" + Arrays.toString(indices) + ", index=" + jdbcUrl.getCatalog() + ", create it");
            transportClient.admin().indices().prepareCreate(jdbcUrl.getCatalog()).get();
        }

        return new BudoElasticsearchJdbcConnection() //
                .setTransportClient(transportClient) //
                .statementParser(STATEMENT_PARSER);
    }

    private TransportClient transportClient(String host, Integer port, Map<String, String[]> parameters) {
        Builder builder = Settings.builder();

        for (ESConf conf : ESConf.values()) {
            String value = MapUtil.getFirst(parameters, conf.getKey(), conf.getDefaultValue());
            builder.put(conf.getKey(), value);
        }

        Settings settings = builder.build();

        TransportClient transportClient = TransportClient.builder() //
                .settings(settings) //
                .build();

        InetAddress inetAddress = JavaNetUtil.getInetAddressByName(host);
        InetSocketTransportAddress transportAddress = new InetSocketTransportAddress(inetAddress, port);

        transportClient.addTransportAddress(transportAddress);
        return transportClient;
    }
}